package com.kjlink.widget.monitoring.service;

import com.kjlink.common.base.vo.BooleanWithMsg;
import com.kjlink.widget.monitoring.config.MonitorConfigParams;
import com.kjlink.widget.monitoring.constant.MonitorConstant;
import com.kjlink.widget.monitoring.entity.MonitorEventBatch;
import com.kjlink.widget.monitoring.entity.MonitorRawEvent;
import com.kjlink.widget.monitoring.repository.MonitorEventBatchRepository;
import com.kjlink.widget.monitoring.repository.MonitorRawEventRepository;
import com.kjlink.widget.monitoring.schedule.handle.MonitorEventSendHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * 描述:
 * 事件源消费服务
 *
 * @author Vic.xu
 * @since 2025-08-13 11:24
 */
@Service
public class EventConsumerService {

    private static final Logger logger = LoggerFactory.getLogger(EventConsumerService.class);

    @Resource
    private MonitorConfigParams monitorConfigParams;
    @Resource
    private MonitorRawEventRepository rawEventRepository;

    @Resource
    private MonitorEventBatchRepository batchRepository;

    @Resource
    private StringRedisTemplate stringRedisTemplate;
    @Resource
    private EventBatchService eventBatchService;

    @Resource
    private MonitorEventSendHandler monitorEventSendHandler;


    /**
     * 保存原始事件到数据库
     * 用于接收外部事件并入库
     * 并且会创建或者合并批次
     */
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void saveRawEvent(MonitorRawEvent event) {
        // 生成事件ID（如果未指定）
        if (event.getEventId() == null) {
            event.setEventId(UUID.randomUUID().toString());
        }
        // 检查事件是否已存在（避免重复入库）
        boolean eventExists = rawEventRepository.findByEventId(event.getEventId()).isPresent();
        if (eventExists) {
            logger.warn("Event already exists: {}", event.getEventId());
            return;
        }
        // 保存事件（状态默认为NEW）
        rawEventRepository.save(event);
        logger.info("Saved raw event: {}", event.getEventId());
        // 处理批次
        createOrMergeBatch(event);
    }

    /**
     * 创建或合并批次:
     * <p>
     * 1.  首先根据redis查询上一个批次 ,
     * * 1.1 在时间窗口内则合并 (因为在时间窗口内, 即 时间窗口尚未结束, 不可能触发上传动作, 也就不可能失败)
     * * 1.2 判断当前事件的发生时间 在 批次窗口的开始时间 之前, 表示这个事件 被消费的太迟了,FIXME  是否考虑丢弃
     * * 1.3 当前事件发生时间 在 批次窗口的结束时间 之后, 则根据状态 做不同的操作
     * ** 1.3.1   判断批次已经上传成功, 则创建新的批次
     * ** 1.3.2   批次未上传成功, 则触发一次上报动作
     * *** 1.3.2.1 上报成功: 则创建新的批次
     * *** 1.3.2.2 上报失败: 则 从 这个失败的批次中创建新的批次
     * 2. 如果查不到, 则根据时间窗口查询一次数据库未上报批次(不再额外判断窗口之外的批次!!!) FIXME  是否有必要
     * * 2.1  如果查到批次, 则合并到这个批次
     * 3. 如果没有批次, 则创建新的批次
     * </p>
     */
    public void createOrMergeBatch(MonitorRawEvent event) {
        // 1. 从redis 查询出上一个创建的批次
        MonitorEventBatch lastCreatedBatch = findLastCreatedBatch();
        if (lastCreatedBatch != null) {
            boolean processEventByRedisBatchResult = processEventByRedisBatch(event, lastCreatedBatch);
            if (processEventByRedisBatchResult) {
                return;
            }
        }
        //  2 如果redis 不存在是根据时间窗口查询? 是否有必要
        if (lastCreatedBatch == null) {
            if (processEventByWindow(event)) {
                return;
            }
        }
        // 3. 创建新的批次
        createNewBatch(event);
    }

    /**
     * 1 根据redis 中 最新的批次 处理 事件
     */
    private boolean processEventByRedisBatch(MonitorRawEvent event, MonitorEventBatch lastCreatedBatch) {
        // 1.1 判断是否在一个时间窗口内: 当前event 的 发生时间 在上一个批次的 时间窗口内 firstOccurTime - endTime
        boolean innerWindow = event.getOccurTime().isBefore(lastCreatedBatch.getEndTime())
                && event.getOccurTime().isAfter(lastCreatedBatch.getFirstOccurTime());
        // 1.1 在时间窗口内则合并 (因为在时间窗口内, 即 时间窗口尚未结束, 不可能触发上传动作, 也就不可能失败)
        if (innerWindow) {
            mergeToBatch(lastCreatedBatch, event);
            return true;
        }
        // 如果不在时间窗口内 判断窗口 和当前事件的时间关系
        boolean isBefore = event.getOccurTime().isBefore(lastCreatedBatch.getFirstOccurTime());
        if (isBefore) {
            // 1.2 判断当前事件的发生时间 在 批次窗口的开始时间 之前, 表示这个事件 被消费的太迟了,FIXME  是否考虑不处理
            logger.warn("Event {} is too late, discarded", event.getEventId());
            return false;
        }
        // 当前事件发生时间 在 批次窗口的结束时间 之后,
        boolean isAfter = event.getOccurTime().isAfter(lastCreatedBatch.getEndTime());
        // 1.3 批次窗口已结束,
        if (isAfter) {
            boolean processEventAfterBatchResult = processEventAfterBatch(event, lastCreatedBatch);
            if (processEventAfterBatchResult) {
                return true;
            }
        }
        return false;
    }

    /**
     * 2 当前事件发生时间 在 批次窗口的结束时间 之后, 则根据状态 做不同的操作
     */
    private boolean processEventAfterBatch(MonitorRawEvent event, MonitorEventBatch lastCreatedBatch) {
        // 1.3.1 判断批次已经上传成功, 则创建新的批次
        if (lastCreatedBatch.getStatus() == MonitorEventBatch.BatchStatus.success) {
            // 创建的新批次
            createNewBatch(event);
            return true;
        }
        // 1.3.2 批次窗口未结束, 则触发一次上报动作
        if (lastCreatedBatch.getStatus() == MonitorEventBatch.BatchStatus.initial) {

            BooleanWithMsg send = monitorEventSendHandler.send(lastCreatedBatch);
            if (send.isSuccess()) {
                // 1.3.2.1 上报成功: 则创建新的批次
                createNewBatch(event);
            } else {
                //  1.3.2.2 失败  则 从 这个失败的批次中创建新的批次
                createNewBatchFromPreFailedBatch(lastCreatedBatch, event);
            }
            return true;
        }
        return false;

    }

    /**
     * 3 如果redis 不存在是根据时间窗口查询? FIXME 是否有必要
     */
    private boolean processEventByWindow(MonitorRawEvent event) {
        MonitorEventBatch lastCreatedBatch = batchRepository.findFirstPendingBatch(event.getEventType(),
                event.getOccurTime(),
                MonitorEventBatch.BatchStatus.initial
        ).orElse(null);
        if (lastCreatedBatch != null) {
            mergeToBatch(lastCreatedBatch, event);
            return true;
        }
        return false;
    }

    protected MonitorEventBatch buildBatch(MonitorRawEvent event) {
        // 生成批次号（事件类型+时间戳+随机数）
        String batchNo = String.format("%s_%d_%s",
                event.getEventType(),
                System.currentTimeMillis(),
                UUID.randomUUID().toString().substring(0, 8));

        // 创建批次
        MonitorEventBatch batch = new MonitorEventBatch();
        batch.setBatchNo(batchNo);
        batch.setEventType(event.getEventType());
        batch.setBusinessId(event.getBusinessId());
        batch.setFirstOccurTime(event.getOccurTime());
        batch.setLastOccurTime(event.getOccurTime());
        batch.setCreatedBy(event.getCreatedBy());
        // 设置截止时间
        batch.setEndTime(event.getOccurTime().plus(monitorConfigParams.getBatchWindowMinutes(), ChronoUnit.MINUTES));
        batch.setFirstEventId(event.getId());
        batch.setEventCount(1);
        batch.setLatestData(event.getData());
        return batch;
    }

    /**
     * 创建新批次
     *
     * @return 批次 id
     */
    public void createNewBatch(MonitorRawEvent event) {

        // 创建批次
        MonitorEventBatch batch = buildBatch(event);
        batchRepository.save(batch);
        // 记录最新创建的批次的id
        recordLastCreatedBatch(batch);

        // 更新事件批次号
        event.setBatchNo(batch.getBatchNo());
        rawEventRepository.save(event);

        logger.info("Created new batch {} for event {}", batch.getBatchNo(), event.getEventId());
    }


    /**
     * 记录最新创建的批次的id
     */
    public void recordLastCreatedBatch(MonitorEventBatch batch) {
        stringRedisTemplate.opsForValue().set(MonitorConstant.LAST_CREATED_BATCH_KEY, batch.getId(), monitorConfigParams.getExpireMinutes(), TimeUnit.MINUTES);
    }

    /**
     * 获取最新创建的批次的id
     */
    public String getLastCreatedBatchId() {
        return stringRedisTemplate.opsForValue().get(MonitorConstant.LAST_CREATED_BATCH_KEY);
    }

    /**
     * 获取最后创建的批次
     */
    public MonitorEventBatch findLastCreatedBatch() {
        String batchId = getLastCreatedBatchId();
        if (batchId == null) {
            return null;
        }
        return batchRepository.findById(batchId).orElse(null);
    }

    /**
     * 合并事件到已有批次
     */
    public void mergeToBatch(MonitorEventBatch batch, MonitorRawEvent event) {
        // 更新批次信息
        batch.setLastOccurTime(event.getOccurTime());
        batch.setEventCount(batch.getEventCount() + 1);
        // 保留最新数据
        batch.setLatestData(event.getData());
        batchRepository.save(batch);

        // 更新事件状态和批次号
        event.setBatchNo(batch.getBatchNo());
        rawEventRepository.save(event);
        logger.info("Merged event {} to batch {}", event.getEventId(), batch.getBatchNo());
    }


    /**
     * 处理单个事件的批次：判断是否加入已有批次或创建新批次
     */
    @Deprecated
    private void processEventBatch(MonitorRawEvent event) {
        // 查询此类型 是否有未上送的批次
        MonitorEventBatch existingBatch = batchRepository.findFirstPendingBatch(event.getEventType(),
                event.getOccurTime(),
                MonitorEventBatch.BatchStatus.initial
        ).orElse(null);

        // 当前批次存在的时候,合并到该批次
        if (existingBatch != null) {
            // 已有批次：检查是否在窗口内
            if (event.getOccurTime().isBefore(existingBatch.getEndTime())) {
                // 在窗口内：合并到该批次
                mergeToBatch(existingBatch, event);
                return;
            }
        }
        //  FIXME 当前批次不存在的时候, 查询出上一个压缩窗口失败的批次   redis 记录上一个批次号,  此处查询出批次 判断是否是失败的, 这样就不用根据时间窗口查询(不连续的会查不到)
        MonitorEventBatch preFailedBatch = findPreFailedBatch(event).orElse(null);
        // 创建新的批次, 但是把上一次失败的批次 压缩次数累加进来
        if (preFailedBatch != null) {
            createNewBatchFromPreFailedBatch(preFailedBatch, event);
            return;
        }
        // 无批次或超出窗口：创建新批次
        createNewBatch(event);
    }

    /**
     * 创建新的批次
     *
     * @param preFailedBatch
     * @param event
     */
    @Deprecated
    protected void createNewBatchFromPreFailedBatch(MonitorEventBatch preFailedBatch, MonitorRawEvent event) {
        MonitorEventBatch batch = buildBatch(event);
        batch.setLegacyBatchFrom(preFailedBatch.getId());
        batch.setLegacyBatchCount(preFailedBatch.getCount());
        // 需要把 事件发生时间改为上一个批次的 时间首次发生时间呢?  先这样写,后面根据需求再决定是否改动
        batch.setFirstOccurTime(preFailedBatch.getFirstOccurTime());
        batch.setFirstEventId(preFailedBatch.getFirstEventId());
        batch.setBusinessId(preFailedBatch.getBusinessId());
        batchRepository.save(batch);
        // 记录最新创建的批次的id
        recordLastCreatedBatch(batch);

        // 更新事件批次号
        event.setBatchNo(batch.getBatchNo());
        rawEventRepository.save(event);

        logger.info("Created new batch {} for preFailedBatch and event  {}", batch.getBatchNo(), event.getEventId());
    }

    /**
     * 查询出上一个压缩窗口失败的批次
     */
    @Deprecated
    public Optional<MonitorEventBatch> findPreFailedBatch(MonitorRawEvent event) {
        LocalDateTime occurTime = event.getOccurTime();
        // 把时间减少一个压缩窗口, 已查询上一个失败批次
        LocalDateTime windowStart = occurTime.minus(monitorConfigParams.getBatchWindowMinutes(), ChronoUnit.MINUTES);
        return batchRepository.findFirstPendingBatch(event.getEventType(), windowStart, MonitorEventBatch.BatchStatus.fail);
    }


}
